package com.funny.rabbitmq.confirm;

import com.funny.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class Consumer {

    private final static String QUEUE_NAME = "hello";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtils.getChannel();

        // 1: 不公平分发hello
//        channel.basicQos(1);
        // 预取值
//        channel.basicQos( 2);
        /**
         * param1：prefetchSize，消息本身的大小 如果设置为0 那么表示对消息本身的大小不限制
         * param2：prefetchCount，告诉rabbitmq不要一次性给消费者推送大于N个消息
         * param3：global，是否将上面的设置应用于整个通道
         * false：表示只应用于当前消费者
         * true：表示当前通道的所有消费者都应用这个限流策略
         */
        // 限流
        channel.basicQos(0, 5, false);
        /**
         * 消费者消费参数
         * 1.消费队列名称
         * 2.是否自动应答
         * 3.消费者未成功消费的回调
         * 4.消费者取消消费的回调
         */
        channel.basicConsume(QUEUE_NAME, false, (consumerTag, message) -> {
            String msg = new String(message.getBody());
            log.info("consumerTag={}, message={}", consumerTag, msg);
            boolean result = handleMsg(msg);
            long deliveryTag = message.getEnvelope().getDeliveryTag();
            log.info("deliveryTag={}", deliveryTag);
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            if (result) {
                channel.basicAck(deliveryTag, false);
            } else {
                channel.basicNack(deliveryTag, false, true);
//                channel.basicReject(deliveryTag, false);
            }
        }, consumerTag -> {
            log.info("consumerTag={}", consumerTag);
        });
        System.out.println("等待消费...");

    }

    private static boolean handleMsg(String msg) {
        String[] strs = msg.split("_");
        int n = Integer.parseInt(strs[1]);
        return n > 2;
    }
}
